package com.atguigu.flink.chapter05.transform;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/*
Union :
    1、合并多个流
    2、流的类型必须一致
 */
public class UnionDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port",2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        DataStreamSource<Integer> stream = env.fromElements(1, 2, 3, 4, 5);
        DataStreamSource<Integer> stream2 = env.fromElements(23, 24, 25, 26, 27);
        DataStreamSource<Integer> stream3 = env.fromElements(31, 32, 33, 34, 35);
        DataStream<Integer> cs = stream.union(stream2, stream3);
        SingleOutputStreamOperator<String> s1 = cs.map(new MapFunction<Integer, String>() {
            @Override
            public String map(Integer value) throws Exception {
                return value + "-->";
            }
        });

        cs.print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}


/*
union:
    1、合并多个流
    2、流的类型必须一致
 */
//public class UnionDemo {
//    public static void main(String[] args) {
//        Configuration conf = new Configuration();
//        conf.setInteger("rest.port",2000);
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
//        env.setParallelism(1);
//        DataStreamSource<String> stream = env.fromElements("a", "b", "c", "d");
//        DataStreamSource<String> stream1 = env.fromElements("A", "B", "C", "D");
//        DataStreamSource<String> stream2 = env.fromElements("a1", "b1", "c1", "d1");
//        DataStream<String> s1 = stream.union(stream1, stream2);
//        s1.print();
//
//        try {
//            env.execute();
//        } catch (Exception e) {
//            e.printStackTrace();
//        }
//    }
//}